package com.example.liucxu.myapplication;


import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.view.View;
import android.widget.TextView;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import butterknife.BindView;
import butterknife.ButterKnife;
import rx.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.android.schedulers.AndroidSchedulers;
import rx.functions.Action1;
import rx.functions.Func0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.schedulers.Schedulers;

public class MainActivity extends AppCompatActivity {
    @BindView(R.id.start_tv)
    TextView start_tv;
    @BindView(R.id.timer_tv)
    TextView timer_tv;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        ButterKnife.bind(this);
        final ArrayList<String> sources = new ArrayList<>();
        for (int index = 0; index < 7; index++) {
            sources.add(index + "");
        }
        //buffer操作符 缓存数据然后发送
        Observable.just("0", "1", "2", "3", "4", "5", "6").buffer(2, 3).subscribe(new Subscriber<List<String>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(List<String> strings) {
                Log.i("onNext", strings.toString());
            }
        });
        Observable.from(sources).buffer(2, 3).subscribe(new Subscriber<List<String>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(List<String> strings) {
                Log.i("onNext", strings.toString());
            }
        });

        //concat 操作符 连接两个Observable
        Observable.concat(Observable.just("0", "1", "2", "3", "4", "5", "6"), Observable.just("7", "8")).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                Log.i("onNext", s);
            }
        });

        //defer 数据源发生变化了之后，第二个订阅的Observer会得到不同的数据
        Observable<String> defer_observable = Observable.defer(new Func0<Observable<String>>() {
            @Override
            public Observable<String> call() {
                return Observable.from(sources);
            }
        });
        sources.add("3");
        defer_observable.subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                Log.i("onNext", s);
            }
        });
        //distinct 去重
        Observable.from(sources).distinct().subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                Log.i("onNext", s);
            }
        });
        //filter 过滤
        Observable.from(sources).filter(new Func1<String, Boolean>() {
            @Override
            public Boolean call(String s) {
                if (Integer.parseInt(s) > 3) {
                    return true;
                } else
                    return false;
            }
        }).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                Log.i("onNext", s);
            }
        });

        //reduce 对所有数据进行处理，最终emit一个数据
        Observable.from(sources).reduce(new Func2<String, String, String>() {
            @Override
            public String call(String s, String s2) {
                return s + s2;
            }
        }).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(String s) {
                Log.i("onNext", s);
            }
        });
        //interval interval可以延时一定时间后开始按周期emit数据，emit的数据从0开始一次递增。
        Observable<Long> timerCount;
        final int time = 10;
        timerCount = Observable.interval(0, 1, TimeUnit.SECONDS).subscribeOn(AndroidSchedulers.mainThread())
                .observeOn(AndroidSchedulers.mainThread());
        final Subscription subscription;
        subscription = timerCount.map(new Func1<Long, Integer>() {
            @Override
            public Integer call(Long aLong) {
                Log.i("onNext", aLong + "");
                return time - aLong.intValue();
            }
        }).take(time + 1).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                timer_tv.setText(integer + "");
            }
        });
        start_tv.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
            }
        });

        //scan
        Observable.just(3, 7, 15, 9).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).scan(new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) {
                return integer + integer2;
            }
        }).subscribe(new Action1<Integer>() {
            @Override
            public void call(Integer integer) {
                Log.i("onNext", integer + "");
            }
        });
    }
}
